//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////// FOR READING FROM FILE ///////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

package trunk.hadoop;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.io.BytesWritable;

import java.io.Closeable;
import java.io.IOException;

public class daInputFileFetcher implements Closeable {
    private FSDataInputStream inputStream;
    private int     longestQuery,  blockSize;
    private int     posRelativeToSplit = 0;
    private byte[]  byteInput;

    public daInputFileFetcher(FSDataInputStream is, long blockStartOffset, long size, int lQuery, int bSize){
        inputStream = is;
        blockSize = bSize;
        longestQuery = lQuery;

        try{    // allocate input in to mem

            byteInput = new byte[(int)size];           // size ของ [] ที่โยนให้ map = size ของไฟล์ split

            inputStream.seek(blockStartOffset);  // jump to preferred position

            for (int i = 0; i < size; i++){                    // puts input into mem at set size
                if (inputStream.read(byteInput, i,1) == -1) break;  // read into mem and in case of unexpected eof, end loop
            }
            // TODO : to check if it read properly
            // System.out.println("source data = " + new String(byteInput)); //System.out.println("Data read from stream: " + i);

        } catch (IOException e){ e.printStackTrace(); }
    }

    public int read(BytesWritable valueIn) throws IOException{
        if (byteInput.length - posRelativeToSplit > blockSize) {
            //System.out.println("Pos " + posRelativeToSplit + "-" + (posRelativeToSplit + splitSize));
            valueIn.set(byteInput, posRelativeToSplit, blockSize);          // โยน byte[]กลับไปให้map       set(byte[] newData, int offset, int length
            posRelativeToSplit += (blockSize - longestQuery);                                  // set how much to jump
        }
        else
        {
            //System.out.println("Pos " + posRelativeToSplit + "-" + (posRelativeToSplit + (byteInput.length - posRelativeToSplit)));
            valueIn.set(byteInput, posRelativeToSplit, byteInput.length - posRelativeToSplit);          // for the end of buffer
            //System.out.println(new String(valueIn.getBytes()));
            return -1;
        }
        //System.out.println(new String(valueIn.getBytes()));
        return 0;

    }
    @Override
    public void close() throws IOException { inputStream.close(); }
}
